package com.atbeijing.D06;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 双流connect 实时对账
 */
public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> zhifuStream = env
                .fromElements(
                        new Event("order-1", "zhifu", 1000L),
                        new Event("order-2", "zhifu", 2000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        SingleOutputStreamOperator<Event> weixinStream = env
                .fromElements(
                        new Event("order-1", "weixin", 3000L),
                        new Event("order-3", "weixin", 4000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );
        zhifuStream
                .keyBy(r -> r.orderId)
                .connect(weixinStream.keyBy(r->r.orderId))
                //订单id相同数据分到一条流用一个process处理
                .process(new myCoProcess())
                .print();



        env.execute();
    }


    //app数据与微信数据,谁先到谁先被处理,然后等待对方一定时间
    public static class myCoProcess extends CoProcessFunction<Event,Event,String>{
       //如果没有匹配成功暂时保存,过了保存时间就下发,告知没有成功支付
        private ValueState<Event> zhifu;
        private ValueState<Event> weixin;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            zhifu=getRuntimeContext().getState(new ValueStateDescriptor<Event>("zhi-fu",Types.POJO(Event.class)));
            weixin=getRuntimeContext().getState(new ValueStateDescriptor<Event>("wei-xin",Types.POJO(Event.class)));
        }

        //app支付数据到来执行
        @Override
        public void processElement1(Event value, Context ctx, Collector<String> out) throws Exception {
            //微信数据已经收集到
            if (weixin.value()!=null){
                out.collect("订单ID " + value.orderId + "匹配成功");
                weixin.clear();
            }else {
                //微信数据没有匹配到,将app数据存起来,并等待5s时间
                zhifu.update(value);
                ctx.timerService().registerEventTimeTimer(5000L);
            }

        }

        //微信支付数据到来执行
        @Override
        public void processElement2(Event value, Context ctx, Collector<String> out) throws Exception {
            //支付数据已经收集到
            if (zhifu.value()!=null){
                out.collect("订单ID " + value.orderId + "匹配成功");
                zhifu.clear();
            }else {
                //支付数据没有匹配到,将微信数据存起来,并等待5s时间
                weixin.update(value);
                ctx.timerService().registerEventTimeTimer(5000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            //说明有app数据没匹配到微信数据
            if (zhifu.value()!=null){
                out.collect(zhifu.value().orderId + "的微信事件没到");
                zhifu.clear();
            }
            //说明有微信数据没匹配到支付数据
            if (weixin.value() != null) {
                out.collect(weixin.value().orderId + "的支付事件没到");
                weixin.clear();
            }
        }
    }
    public static class Event {
        public String orderId;
        public String eventType;
        public Long timestamp;

        public Event() {
        }

        public Event(String orderId, String eventType, Long timestamp) {
            this.orderId = orderId;
            this.eventType = eventType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "orderId='" + orderId + '\'' +
                    ", eventType='" + eventType + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

}
